use std::time::Duration;
use async_trait::async_trait;
use bytes::{Bytes, BytesMut};
use log::{error, info};
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::oneshot;
use rpc::connection_manager::connection::{ConnectionHandle};
use rpc::{decode_rpc_header, encode_rpc_data, ChannelId, RpcChannel, RpcContext, RpcResult};
use rpc::connection_manager::event::{ResponseData, WriteData};
use rpc::dto::{RpcData, RpcHeader};
use rpc::router::RpcRouter;
use rpc::sender::RpcSender;
use crate::connection_manager::message::ConnectionMessage;
use crate::match_express;

pub struct ServerConnectionHandler {
    router: RpcRouter,
    channel: RpcServerChannel,
    sender: RpcSender,
}

impl ServerConnectionHandler {
    pub fn new(router: RpcRouter, connect_id: ChannelId, sender: RpcSender) -> Self {
        let channel = RpcServerChannel::new(connect_id, sender.clone());
        Self {
            router,
            channel,
            sender,
        }
    }

    fn do_handle(&mut self, rpc_context: &mut RpcContext, header: &RpcHeader, data_bytes: Vec<u8>)-> RpcResult<()> {
        let status = header.status;
        if status != 0 { // 消息回执
            let response_data = ResponseData::new(header.local_seq().unwrap(), status != -10, data_bytes);
            let connection_message = ConnectionMessage::Response(response_data);
            self.sender.send_data(Box::new(connection_message))?;
            return Ok(());
        }
        let path = header.path.clone();
        let router_factory = self.router.get_router(path.as_str());

        if let Some(router_factory) = router_factory {
            let mut do_router = (router_factory).parse(rpc_context, BytesMut::from(Bytes::from(data_bytes)))?;

            let filters = self.router.get_filters();
            // 处理过滤器： before
            for (expression, filter) in filters.iter() {
                if *expression == "*" || match_express(&*expression, &path) {
                    filter.before(rpc_context, &mut *do_router)?;
                }
            }

            let executor_result = do_router.execute(rpc_context);

            match executor_result {
                Ok(mut result) => {
                    // 处理过滤器 after
                    for (expression, filter) in filters.iter() {
                        if *expression == "*" || match_express(&*expression, &path) {
                            filter.after(rpc_context, &mut * result)?;
                        }
                    }

                    let local_seq = header.local_seq().unwrap();
                    let response_header = header.response_header(10);
                    let data_bytes = result.encode(rpc_context)?;
                    let rpc_data = RpcData::build(response_header, data_bytes);
                    let bytes = encode_rpc_data(rpc_context, &rpc_data)?;
                    let write_result = self.channel.write_result_no_response(local_seq, bytes);
                    if let Err(e) = write_result {
                        error!("Failed to send response to connection manager, {}", e);
                    }
                }
                Err(error) => {
                    let local_seq = header.local_seq().unwrap();
                    let response_header = header.response_header(-10);
                    let rpc_data_bytes = RpcData::build(response_header.clone(), error);
                    let bytes = encode_rpc_data(rpc_context, &rpc_data_bytes)?;
                    let result = self.channel.write_result_no_response(local_seq, bytes);
                    if let Err(e) = result {
                        error!("Failed to send response to connection manager, {}", e);
                    }
                }
            }
        }

        Ok(())
    }
}

#[async_trait]
impl ConnectionHandle for ServerConnectionHandler {
    async fn connect_open(&mut self, channel_id: ChannelId, sender: UnboundedSender<Bytes>) -> RpcResult<()> {
        info!("router_service connection open: {:?}", channel_id);
        let (oneshot_sender, oneshot_receiver) = oneshot::channel::<bool>();
        let send_result = self.sender.send_data(Box::new(ConnectionMessage::NewConnection {channel_id, sender, recv_sender: oneshot_sender}));
        if let Err(err) = send_result {
            panic!("Failed to connect to connection manager");
        }
        oneshot_receiver.await?;
        Ok(())
    }

    fn connect_clone(&mut self, id: ChannelId) {
        let send_result = self.sender.send_data(Box::new(ConnectionMessage::Disconnection(id)));
        if let Err(err) = send_result {
            panic!("Failed to connect to connection manager");
        }
    }

    fn handle(&mut self, id: ChannelId, rpc_context: &mut RpcContext, bytes: Vec<u8>) -> RpcResult<()> {
        let (header, data_bytes) = decode_rpc_header(rpc_context, &bytes)?;
        let result = self.do_handle(rpc_context, &header, data_bytes);
        if let Err(err) = result {
            let rpc_data = RpcData::build(header.response_header(-10), err);
            let datas = encode_rpc_data(rpc_context, &rpc_data)?;
            let write_data = WriteData::new(id, header.local_seq().unwrap(), datas, false, None);
            let message = ConnectionMessage::Write(write_data);
            rpc_context.send_data(Box::new(message))?;
        }
        Ok(())
    }
}


#[derive(Clone)]
pub struct RpcServerChannel {
    connection_id: ChannelId,
    sender: RpcSender,
}

impl RpcServerChannel {
    pub fn new(connection_id: ChannelId, sender: RpcSender) -> Self {
        Self { connection_id,  sender }
    }

    pub fn write_result_no_response(&mut self, local_seq: String, data: Vec<u8>)-> RpcResult<()> {
        // info!("write result no response, {:?}", rpc_data_bytes.header);
        let write_data = WriteData::new(self.connection_id, local_seq, data, false, None);
        let event = ConnectionMessage::Write(write_data);
        self.sender.send_data(Box::new(event))?;
        Ok(())
    }
}

#[async_trait]
impl RpcChannel for RpcServerChannel {
    async fn write(&mut self, local_seq: String, data: Vec<u8>, timeout: u32) -> RpcResult<Vec<u8>> {
        let (sender, receive) = oneshot::channel();
        let write_data = WriteData::new(self.connection_id, local_seq, data, true, Some(sender));
        let connection_message = ConnectionMessage::Write(write_data);
        self.sender.send_data(Box::new(connection_message))?;
        let data = tokio::time::timeout(Duration::from_millis((timeout * 100) as u64), receive).await;
        match data {
            Ok(data) => {
                match data {
                    Ok(data) => {
                        Ok(data.data)
                    }
                    Err(error) => {
                        error!("Failed to write to connection manager, {}", error);
                        Err(error.into())
                    }
                }

            }
            Err(error) => {
                info!("router_service send data timeout: {:?}", error);
                Err(error.into())
            }
        }

    }
}